package me.yuzho.study.concurrency;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

import static net.mindview.util.Print.*;

class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
    
    private static Random rand = new Random(47);
    
    private static int counter = 0;
    private final int id = counter++;
    
    private final int priority;
    
    protected static List<PrioritizedTask> sequence = new ArrayList<PrioritizedTask>();
    
    public PrioritizedTask(int priority) {
        this.priority = priority;
        sequence.add(this);
    }

    @Override
    public String toString() {
        return String.format("[%1$-3d]", priority) + " Task " + id;
    }
    
    public String summary() {
        return "(" + id + " : " + priority + ")";
    }
    
    @Override
    public int compareTo(PrioritizedTask o) {
        PrioritizedTask that = o;
        return this.priority < that.priority? 1 : 
            (this.priority > that.priority? -1 : 0);
    }

    @Override
    public void run() {
        try {
            TimeUnit.MILLISECONDS.sleep(rand.nextInt(250));
        } catch (InterruptedException e) {
            print(e + ": accepatance way to exit");
        }
        
        print(this);
    }
    
    
    public static class EndSentinel extends PrioritizedTask {

        private ExecutorService exec;
        
        public EndSentinel(ExecutorService e) {
            super(-1); // lowest priority in this program
            exec = e;
        }
        
        @Override
        public void run() {
            int count = 0;
            for(PrioritizedTask pt : sequence) {
                printnb(pt.summary());
                if(++count % 5 == 0) {
                    print();
                }
            }
            print();
            print(this + " Calling shutdownNow()");
            exec.shutdownNow();
        }
        
    }
}

class PrioritizedTaskProducer implements Runnable {

    private static Random rand = new Random(47);
    
    private Queue<Runnable> queue;
    
    private ExecutorService exec;
    
    public PrioritizedTaskProducer(Queue<Runnable> q, ExecutorService e) {
        queue = q;
        exec = e;
    }
    
    @Override
    public void run() {
        // Unbounded queue, never blocks, fill it up fast with random priorities
       for(int i=0; i<20; i++) {
           queue.add(new PrioritizedTask(rand.nextInt(10)));
           Thread.yield();
       } 
       
       // Trickle in highest-priority jobs
       try {
           for(int i=0; i < 10; i++) {
               TimeUnit.MILLISECONDS.sleep(250);
               queue.add(new PrioritizedTask(10));
           }
           
           // add jobs, lowest priority first:
           for(int i=0; i < 10; i++) {
               queue.add(new PrioritizedTask(i));
           }
           
           
           // A sentinel to stop all tasks
           queue.add(new PrioritizedTask.EndSentinel(exec)); 
        
        } catch (InterruptedException e) {
            print("Producer interrupted when sleep");
        }
       
        print("Finished PrioritizedTaskProducer");
    }
    
}


class PrioritizedTaskConsumer implements Runnable {
    
    private PriorityBlockingQueue<Runnable> q;
    
    public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> q) {
        this.q = q;
    }
    
    @Override
    public void run() {
        try {
            while (!Thread.interrupted()) {
                q.take().run(); // Use current thread to run the task
            }
            
        } catch (InterruptedException e) {
            print("Interruptted when taking a task form queue");
        }
        print("Finished PrioritizedTaskConsumer");
    }
}

public class PriorityBlockingQueueDemo {

    public static void main(String[] args) {
        //Random rand = new Random(47);
        
        ExecutorService exec = Executors.newCachedThreadPool();
        PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<Runnable>();
        
        exec.execute(new PrioritizedTaskProducer(queue, exec));
        exec.execute(new PrioritizedTaskConsumer(queue));
    }
}
